package goredis

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/Shopify/sarama"
	uuid "github.com/satori/go.uuid"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/astaxie/beego/logs"
	"github.com/go-redis/redis/v8"
)

// Cache 先构建一个Cache实例，然后将配置参数传入该实例的StartAndGC方法来初始化实例和程序进程退出后的清理工作。
type Cache struct {
	client     *redis.ClusterClient
	prefix     string
	marshal    func(v interface{}) ([]byte, error)
	unmarshal  func(data []byte, v interface{}) error
	consistent map[string]*Consistent
}

// New 根据配置参数创建redis工具实例
func New(options *redis.ClusterOptions, prefix ...string) (*Cache, error) {
	c := new(Cache)
	c.client = redis.NewClusterClient(options)
	c.unmarshal = json.Unmarshal
	c.marshal = json.Marshal
	c.consistent = make(map[string]*Consistent, 0)
	if len(prefix) != 0 {
		c.prefix = prefix[0]
	}
	return c, c.client.Ping(c.client.Context()).Err()
}

/*stream 发布订阅*/

type QueueInterface interface {
	Produce(topic, key string, data interface{}) error
	Subscribe(ctx context.Context, group string, topics []string, handler ConsumerHandler) error
}

//ConsumerHandler 信息接收处理接口
type ConsumerHandler interface {
	Handle(msg *sarama.ConsumerMessage) error
}

const (
	//SequentialPattern 顺序模式，保证消息顺序性，无消费组，以一个stream作为一个topic的分区，
	//分区数量可自定义，一个分区同一时间只能被一个消费者获取，同一标识消息只会汇入同一分区 .
	SequentialPattern = "SequentialPattern"
	//NativePattern 原生模式，group 组之间存在竞争.
	NativePattern = "NativePattern"
)

//Queue 默认使用原生模式
func (c *Cache) Queue(pattern string) QueueInterface {
	switch pattern {
	case NativePattern:
		return new(nativeQueue)
	case SequentialPattern:
		return new(sequentialQueue)
	default:
		return new(nativeQueue)
	}
}

type nativeQueue struct {
	*Cache
}

// Produce .
func (c *nativeQueue) Produce(topic, key string, data interface{}) error {
	//创建一个结构体并赋值
	d := &struct {
		UUID uuid.UUID   `json:"uuid"`
		Data interface{} `json:"data"`
	}{
		Data: data,
		UUID: uuid.NewV4(),
	}
	//json序列化
	value, err := json.Marshal(d)
	if err != nil {
		return err
	}
	//sendMessage结构体赋值
	msg := &sarama.ProducerMessage{
		Topic:     topic,
		Key:       sarama.ByteEncoder(key),
		Value:     sarama.ByteEncoder(value),
		Timestamp: time.Now(),
	}
	c.client.XAdd(c.client.Context(), &redis.XAddArgs{
		Stream: topic,
		ID:     "*",
		Values: msg,
	})
	return err
}

func (c *nativeQueue) Subscribe(ctx context.Context, group string, topics []string, handler ConsumerHandler) error {
	for _, topic := range topics {
		//只获取最新消息
		result, err := c.client.XGroupCreateMkStream(ctx, topic, group, "$").Result()
		if err != nil {
			return err
		}
		logs.Info("redis topic[%s] group[%s] create success:%s", topic, group, result)
		messages := make(chan *sarama.ConsumerMessage, 2048)
		go func(ctx context.Context, messages chan *sarama.ConsumerMessage) {
			for {
				select {
				case <-ctx.Done():
					return
				case msg := <-messages:
					if err := handler.Handle(msg); err != nil {
						logs.Error("consumer Handle Error:[%s],msg:%s:/%d/%s\t%s ", err.Error(), msg.Topic, msg.Partition, msg.Key, msg.Value)
					}
				}
			}
		}(ctx, messages)
		go func(ctx context.Context, topic, group string, messages chan *sarama.ConsumerMessage) {
			for {
				select {
				case <-ctx.Done():
					return
				default:
					xStreams, err := c.client.XReadGroup(ctx, &redis.XReadGroupArgs{
						Group:    group,
						Consumer: fmt.Sprintf("%s-%s", group, topic),
						Streams:  []string{topic},
						Count:    1,     //指定读取消息的数量
						Block:    3 * time.Second,    //阻塞模式
						NoAck:    false, //不自动提交，等待消息确认，为true即为自动提交，概率性消息丢失
					}).Result()
					if err != nil {
						logs.Error(err.Error())
						continue
					}
					for _, xStream := range xStreams {
						msgsID := make([]string, 0)
						for _, msg := range xStream.Messages {
							val, err := json.Marshal(msg.Values)
							if err != nil {
								logs.Error(err.Error())
								continue
							}
							messages <- &sarama.ConsumerMessage{
								Timestamp:      time.Now(),
								BlockTimestamp: time.Now(),
								Key:            []byte(msg.ID),
								Value:          val,
								Topic:          xStream.Stream,
							}
							msgsID = append(msgsID, msg.ID)
						}
						//消息确认提交
						c.client.XAck(ctx, xStream.Stream, group, msgsID...)
					}
				}
			}
		}(ctx, topic, group, messages)
	}
	return nil
}

type sequentialQueue struct {
	*Cache
}

const queuePartitionNum = 3 //分区数量

// InitStream 根据queuePartitionNum指定每个topic的分区数量，多节点部署需要确保queuePartitionNum一致，一个分区就是一个stream.
func (c *sequentialQueue) initStream(topics []string) {
	for _, topic := range topics {
		if _, ok := c.consistent[topic]; ok {
			continue
		}
		consistent := NewConsistent()
		for i := 0; i < queuePartitionNum; i++ {
			streamKey := fmt.Sprintf("%s-partition%d", topic, i)
			consistent.Add(streamKey)
		}
		c.consistent[topic] = consistent
	}
}

// Produce 根据key做一致性hash操作，同一key类型消息发布顺序发布到同一stream(即同一分区).
func (c *sequentialQueue) Produce(topic, key string, data interface{}) error {
	//创建一个结构体并赋值
	d := &struct {
		UUID uuid.UUID   `json:"uuid"`
		Data interface{} `json:"data"`
	}{
		Data: data,
		UUID: uuid.NewV4(),
	}
	c.initStream([]string{topic})
	//json序列化
	value, err := json.Marshal(d)
	if err != nil {
		return err
	}
	//sendMessage结构体赋值
	msg := &sarama.ProducerMessage{
		Topic:     topic,
		Key:       sarama.ByteEncoder(key),
		Value:     sarama.ByteEncoder(value),
		Timestamp: time.Now(),
	}
	if _, ok := c.consistent[topic]; !ok {
		return errors.New("topic does not specify a partition")
	}
	streamKey, err := c.consistent[topic].Get(key)
	if err != nil {
		return err
	}
	c.client.XAdd(c.client.Context(), &redis.XAddArgs{
		Stream: streamKey,
		ID:     "*",
		Values: msg,
	})
	return err
}

// Subscribe 订阅给定的一个或多个频道的信息，通过竞争锁来获取某个topic中的一个分区，
// 已被获取的分区不会被重复获取，一个分区获取状态由redis自身维护.
func (c *sequentialQueue) Subscribe(ctx context.Context, _ string, topics []string, handler ConsumerHandler) error {
	c.initStream(topics)
	for {
		ok, err := c.Lock("stream-key", time.Now(), 30)
		if err != nil {
			return err
		}
		if !ok {
			continue
		}
		//将被订阅的stream(分区)集合
		streamKeys := make([]string, 0)
		for _, topic := range topics {
			if _, ok := c.consistent[topic]; !ok {
				return fmt.Errorf("topic[%s] does not specify a partition", topic)
			}
			//遍历topic下的分区
			for _, streamKey := range c.consistent[topic].Members() {
				//该分区是否被订阅
				exists, err := c.Exists(streamKey)
				if err != nil {
					return err
				}
				if !exists {
					_ = c.Set(streamKey, time.Now(), 300)
					streamKeys = append(streamKeys, streamKey)
					break
				}
			}
		}
		//从stream中读取消息
		c.streamRead(ctx, handler, streamKeys...)
		c.UnLock("stream-key")
		time.Sleep(3 * time.Second)
	}
}

func (c *sequentialQueue) streamRead(parentCtx context.Context, handler ConsumerHandler, streamKeys ...string) {
	for _, key := range streamKeys {
		ctx, cancelFunc := context.WithCancel(parentCtx)
		go func(ctx context.Context, key string, cancelFunc context.CancelFunc) {
			for {
				select {
				case <-ctx.Done():
					return
				default:
					for i := 0; i < 5; i++ {
						if err := c.Set(key, time.Now(), 15); err != nil {
							//维持分区状态，失败5次将判定该节点失去控制权，重新将该分区加入竞争序列
							if i == 5 {
								logs.Error(err.Error())
								cancelFunc()
								_ = c.Del(key)
								return
							}
							time.Sleep(2 * time.Second)
							continue
						}
						break
					}
				}
				time.Sleep(10 * time.Second)
			}
		}(ctx, key, cancelFunc)
		messages := make(chan *sarama.ConsumerMessage, 2048)
		go func(ctx context.Context, messages chan *sarama.ConsumerMessage) {
			for {
				select {
				case <-ctx.Done():
					return
				case msg := <-messages:
					if err := handler.Handle(msg); err != nil {
						logs.Error("consumer Handle Error:[%s],msg:%s:/%d/%s\t%s ", err.Error(), msg.Topic, msg.Partition, msg.Key, msg.Value)
					}
				}
			}
		}(ctx, messages)
		go func(ctx context.Context, key string, messages chan *sarama.ConsumerMessage) {
			for {
				select {
				case <-ctx.Done():
					return
				default:
					//阻塞模式读（当有新消息时解除堵塞）
					result, err := c.client.XRead(ctx, &redis.XReadArgs{
						Streams: []string{key},
						Count:   1,               //每次读取一条
						Block:   3 * time.Second, //无消息阻塞时间3秒,有新消息立马停止阻塞，3秒后将返回一条空消息
					}).Result()
					if err != nil {
						logs.Error(err.Error())
					}
					for _, res := range result {
						split := strings.SplitN(res.Stream, "-partition", 2)
						if len(split) != 2 {
							logs.Error("partition parse error")
							continue
						}
						for _, msg := range res.Messages {
							val, err := json.Marshal(msg.Values)
							if err != nil {
								logs.Error(err.Error())
								continue
							}
							partition, err := strconv.Atoi(split[1])
							if err != nil {
								logs.Error(err.Error())
								continue
							}
							messages <- &sarama.ConsumerMessage{
								Timestamp:      time.Now(),
								BlockTimestamp: time.Now(),
								Key:            []byte(msg.ID),
								Value:          val,
								Topic:          split[0],
								Partition:      int32(partition),
							}
						}
					}
				}
			}
		}(ctx, key, messages)
	}
}

// Do 执行redis命令并返回结果。执行时从连接池获取连接并在执行完命令后关闭连接。
func (c *Cache) Do(args ...interface{}) (reply interface{}, err error) {
	return c.client.Do(c.client.Context(), args...).Result()
}

// Set 存并设置有效时长。时长的单位为秒。
// 基础类型直接保存，其他用json.Marshal后转成string保存。
func (c *Cache) Set(key string, val interface{}, expire int64) error {
	value, err := c.encode(val)
	if err != nil {
		return err
	}
	if expire > 0 {
		_, err := c.Do("SETEX", c.getKey(key), expire, value)
		return err
	}
	_, err = c.Do("SET", c.getKey(key), value)
	return err
}

// Atomic operation
func (c *Cache) SetNX(key string, val interface{}, expire time.Duration) (bool, error) {
	value, err := c.encode(val)
	if err != nil {
		return false, err
	}

	return c.client.SetNX(c.client.Context(), c.getKey(key), value, expire).Result()
}

var mutex sync.Mutex

//Lock ...
func (c *Cache) Lock(key string, value interface{}, expire int64) (bool, error) {
	mutex.Lock()
	defer mutex.Unlock()
	val, err := c.encode(value)
	if err != nil {
		return false, err
	}
	//logs.Info("time.Duration(expire)*time.Second:", time.Duration(expire)*time.Second)
	return c.client.SetNX(c.client.Context(), c.getKey(key), val, time.Duration(expire)*time.Second).Result()

}

//UnLock ...
func (c *Cache) UnLock(key string) bool {
	mutex.Lock()
	defer mutex.Unlock()
	err := c.Del(key)
	if err != nil {
		return false
	}
	return true
}

// Exists 检查键是否存在
func (c *Cache) Exists(key string) (bool, error) {
	return Bool(c.Do("EXISTS", c.getKey(key)))
}

// Del 删除键
func (c *Cache) Del(key string) error {
	_, err := c.Do("DEL", c.getKey(key))
	return err
}

// getKey 将健名加上指定的前缀。
func (c *Cache) getKey(key string) string {
	return c.prefix + key
}

// encode 序列化要保存的值
func (c *Cache) encode(val interface{}) (interface{}, error) {
	var value interface{}
	switch v := val.(type) {
	case string, int, uint, int8, int16, int32, int64, float32, float64, bool:
		value = v
	default:
		b, err := c.marshal(v)
		if err != nil {
			return nil, err
		}
		value = string(b)
	}
	return value, nil
}

// decode 反序列化保存的struct对象
func (c *Cache) Decode(reply interface{}, err error, val interface{}) error {
	str, err := String(reply, err)
	if err != nil {
		return err
	}
	if str == "" {
		return ErrNil
	}
	return c.unmarshal([]byte(str), val)
}

func convArr(reply interface{}) []string {
	var rs []string
	arrs, ok := reply.([]interface{})
	if ok {
		for _, item := range arrs {
			s, ok := item.(string)
			if ok {
				rs = append(rs, s)
			} else {
				rs = append(rs, "")
			}
		}
	}
	return rs
}
